package org.raymond.iworks.study.redis.redisson;

import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.*;
import org.redisson.config.Config;
import reactor.core.publisher.Mono;

import java.util.concurrent.*;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
public class LockTest {
    // 非公平锁,不保证执行顺序
    public static void main(String[] args) throws Exception {
        test4();
    }

    public static void test1() throws InterruptedException {
        RedissonClient client = client();
        RLock lock = client.getLock("mylock");
        ThreadPoolExecutor pool = new ThreadPoolExecutor(3, 3,
                0, TimeUnit.SECONDS, new LinkedBlockingQueue());
        Runnable r1 = ()->{
            log.info("wait lock...");
            lock.lock();
            log.info("do something");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            lock.unlock();
            log.info("release lock");
        };
        Runnable r2 = ()->{
            log.info("wait lock...");
            lock.lock(1500, TimeUnit.MILLISECONDS);
            log.info("do something");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            lock.unlock();
            log.info("release lock");
        };
        Runnable r3 = ()->{
            try {
                log.info("wait lock...");
                lock.tryLock(3000, 3000, TimeUnit.MILLISECONDS);
                log.info("do something");
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            } finally {
                lock.unlock();
                log.info("release lock");
            }
        };
        pool.execute(r3);
        pool.execute(r3);
        pool.execute(r3);

        log.info("pool shutdown but wait thread");
        pool.shutdown();
        Thread.sleep(3500);
        log.info("shutdown");
        client.shutdown();
    }

    public static void test2() throws InterruptedException, ExecutionException {
        RedissonClient client = client();
        RLock lock = client.getLock("mylock");
        RFuture<Void> lf1 = lock.lockAsync();
        RFuture<Void> lf2 = lock.lockAsync();
        RFuture<Void> lf3 = lock.lockAsync();
        BiConsumer c = (res,ex)->{
            try {
                log.info("do something");
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            log.info("release lock");
            lock.unlockAsync(); // 在main线程中完成
            log.info("release lock end");
        };
        lf1.whenComplete(c);
        lf2.whenComplete(c);
        lf3.whenComplete(c);

        log.info("shutdown");
        client.shutdown();
    }

    // 未掌握
    public static void test3() throws InterruptedException, ExecutionException {
        RedissonClient client = client();
        RLock lock = client.getLock("mylock");
        RFuture<Void> lf1 = lock.lockAsync(5, TimeUnit.SECONDS);
        RFuture<Void> lf2 = lock.lockAsync(5, TimeUnit.SECONDS);
        RFuture<Void> lf3 = lock.lockAsync(5, TimeUnit.SECONDS);
        BiConsumer c = (res,ex)->{
            try {
                log.info("do something");
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            log.info("release lock");
            lock.unlock(); // 为什么没有执行到下一步?
            log.info("release lock end");
        };
        CompletionStage cs1 = lf1.whenCompleteAsync(c);
        CompletionStage cs2 = lf2.whenCompleteAsync(c);
        CompletionStage cs3 = lf3.whenCompleteAsync(c);

        log.info("get");
        lf1.get();
        log.info("to c get");
        lf2.toCompletableFuture().get();
        log.info("to c join");
        lf3.toCompletableFuture().join();

        /*
        log.info("get");
        lf2.toCompletableFuture().get();
        //cs2.toCompletableFuture().get();
        //lf2.toCompletableFuture().get();
        //lf2.toCompletableFuture().join();
        log.info("get");

        //cs3.toCompletableFuture().join();
        //lf3.toCompletableFuture().get();
        //lf3.toCompletableFuture().join();
         */

        log.info("sleep");
        Thread.sleep(10000);
        log.info("shutdown");
        client.shutdown();
    }

    public static void test4(){
        RedissonReactiveClient client = reactiveClient();
        RLockReactive lock = client.getLock("mylock");
        Mono<Void> lockMono = lock.lock();
        Consumer<Void> c = res->{
            try {
                log.info("do something"); // 这里也没有执行
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
        };
        lockMono.doFirst(()->{
            log.info("doOnFirst");
        }).doOnNext(r->{
            log.info("doOnNext");
        }).doOnSuccess(res->{
            log.info("doOnSuccess");
        }).doFinally(res->{
            log.info("doFinally, unlock start");
            lock.unlock();
            // 如果在main线程中,就会在打印block之前执行,如果在redisson-netty线程中,则会在打印block之后执行
            log.info("unlock end");
        }).subscribe(c);

        log.info("block");
        lockMono.block(); // 必须?为什么?

        log.info("shutdown");
        client.shutdown();
        log.info("shutdown end");
    }

    public static RedissonClient client(){
        Config cfg = new Config();
        cfg.useSingleServer().setAddress("redis://192.168.0.125:6379");
        RedissonClient client = Redisson.create(cfg);
        return client;
    }

    public static RedissonReactiveClient reactiveClient(){
        Config cfg = new Config();
        cfg.useSingleServer().setAddress("redis://192.168.0.125:6379");
        RedissonReactiveClient client = Redisson.create(cfg).reactive();
        return client;
    }
}
